package com.swak.consistency.tcc.schedule;

import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;

import com.swak.consistency.tcc.config.TccConfig;
import com.swak.consistency.tcc.context.Phase;
import com.swak.consistency.tcc.context.TccContext;
import com.swak.consistency.tcc.context.TransLog;
import com.swak.consistency.tcc.handle.TccConsistencyHandler;
import com.swak.consistency.tcc.storage.TransLogStorage;
import com.swak.reactivex.threads.Contexts;
import com.swak.reactivex.threads.ScheduledContext;
import com.swak.utils.time.DateTimes;

/**
 * 定时任务 - 定时处理异常的事件
 * 
 * @author DELL
 */
public class TccScheduler implements InitializingBean, ApplicationListener<ApplicationReadyEvent> {
	protected Logger logger = LoggerFactory.getLogger(TccScheduler.class);
	private ScheduledContext scheduledContext;
	private TccConfig config;
	@Autowired
	private TransLogStorage transLogStorage;
	@Autowired
	private TccConsistencyHandler consistencyHandler;

	public TccScheduler(TccConfig config) {
		this.config = config;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		this.scheduledContext = Contexts.createScheduledContext("Tcc-Scheduler-", 1, true, 60, TimeUnit.SECONDS);
	}

	@Override
	public void onApplicationEvent(ApplicationReadyEvent event) {
		scheduledContext.scheduleWithFixedDelay(() -> {
			try {
				this.doTasks();
			} catch (Throwable e) {
				logger.error("TccScheduler Error:{}", e);
			}
		}, config.getTimeout() * 2, config.getTimeout(), TimeUnit.SECONDS);
	}

	private void doTasks() throws Throwable {
		Date delay = new Date(System.currentTimeMillis() - config.getTimeout() * 1000);
		List<TransLog> transLogs = this.transLogStorage.queryAllByDelay(delay);
		for (TransLog trans : transLogs) {
			this.doTask(trans);
		}
	}

	private void doTask(TransLog trans) {
		try {

			if (DateTimes.expired(trans.getCreateTime(), config.getTransTimeout() * 1000)) {
				logger.error("事务已经超过最大的事务超时限制，TransLog:{}", trans);
			}

			// 事务上下文
			TccContext context = new TccContext();
			context.update(trans, Phase.valueOf(trans.getPhase()));

			// Ready 阶段，取消阶段 -- 直接取消
			context.execute(() -> {
				consistencyHandler.retry(context);
				return null;
			});
		} catch (Throwable e) {
			if (trans.getRetryTimes() >= config.getRetryTimes()) {
				logger.error("处理事务次数超过最大重试次数，需要人工介入处理，TransLog:{}", trans);
			}
			transLogStorage.retry(trans, config.getRetryTimes());
		}
	}

}